package com.zhl.spark.core.rddtransform

import com.zhl.spark.core.BasicRdd

/**
 * @program: demos
 * @description: ${description}
 * @author: 刘振华
 * @create: 2020-12-01 10:56
 **/
object AggrateRdd extends BasicRdd {
    def main(args: Array[String]): Unit = {
        test()
    }

    override def logic(): Unit = {
//        sc.makeRDD(List(
//            ("a",-1),("a",-2),("b",-3),
//            ("b",-4),("b",-5),("a",-6)
//        ),2)
//            .aggregateByKey(0)(
//            (x,y)=>Math.max(x,y),
//            (x,y)=>x+y
//        ).collect().foreach(println)

//        sc.makeRDD(List(
//            ("a",-1),("a",-2),("b",-3),
//            ("b",-4),("b",-5),("a",-6)
//        ),2)
//            .aggregateByKey(0)(
//            Math.max(_,_), _+_
//        ).collect().foreach(println)

//        获取出现的平均值

//        sc.makeRDD(List(
//            ("a",1),("a",2),("b",3),
//            ("b",4),("b",5),("a",6)
//        ),2)
//            .aggregateByKey((0,0))(
//                (t,v)=>(t._1+v,t._2+1),
//                (x,y)=>((x._1+y._1)/(x._2+y._2),0)
//        ).map(a=>(a._1,a._2._1)).collect().foreach(println)


//        sc.makeRDD(List(
//            ("a",1),("a",2),("b",3),
//            ("b",4),("b",5),("a",6)
//        ),2)
//            .aggregateByKey((0,0))(
//                (t,v)=>(t._1+v,t._2+1),
//                (x,y)=>(x._1+y._1,x._2+y._2)
//        ).mapValues(s=>s._1/s._2).collect.foreach(println)

//        sc.makeRDD(List(
//            ("a",1),("a",2),("b",3),
//            ("b",4),("b",5),("a",6)
//        ),2)
//            .combineByKey(
//                v=>(v,1),
//                (t:(Int,Int),v)=>(t._1+v,t._2+1),
//                (x:(Int,Int),y:(Int,Int))=>(x._1+y._1,x._2+y._2)
//        ).mapValues(s=>s._1/s._2).collect.foreach(println)

        //分析log日志，时间戳，省份，城市，用户，广告id，空格分隔，统计每个省份，每个广告被点击数量的top3
//        13 山东 潍坊 a1 adverid1
        sc.textFile("spark//datas//agent.log").map(
            line=>{
                val strs = line.split(" ")
                ((strs(1),strs(4)),1)
            }
        ).reduceByKey(_+_)
//            .map(t=>(t._1._1,(t._1._2,t._2)))
            //模式匹配，相当于做了一个检验
            .map{
                case ((a,b),c) =>{
                    (a,(b,c))
                }
            }
            .groupByKey().mapValues(
            v=>{
//                v.toList.sortBy(_._2).reverse.take(3)
                v.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
            }
        )
            .collect().foreach(println)

    }
}
